package com.ibm.cps.servlet.starter;

import java.io.IOException;
import java.util.List;

import org.apache.http.client.ClientProtocolException;
import org.apache.log4j.Logger;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.Lists;
import com.ibm.cps.servlet.HttpClientMonitor;
import com.ibm.cps.servlet.HttpClientMonitor.HTTP_REQUEST_METHOD;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class StormApplicationMonitor implements ProcessMonitor{

	// check detail at https://github.com/Parth-Brahmbhatt/incubator-storm/blob/master/STORM-UI-REST-API.md
	// this class is designed to connect the storm cluster by rest-api and get the info.
	
	private static final Logger logger = Logger.getLogger(StormApplicationMonitor.class);
	private final String resourceManagerURL;
	
	public StormApplicationMonitor(String resourceManagerURL) {
		this.resourceManagerURL = resourceManagerURL;
	}
	
//	/api/v1/topology/summary (GET)
//	Returns summary information for all topologies.
//api/vi/topology/summary
	
	@Override
	public boolean isAlive(String tenantId) throws CPSException {

		try{
			String url = resourceManagerURL + "api/v1/topology/summary";
			logger.info("connecting to storm resource manager: " + url);
			String request = HttpClientMonitor.sendGetRequest(url);
			ObjectNode object = JSONUtility.fromObjectString(request);
			ArrayNode apps = (ArrayNode)object.get("topologies");
			int appCount = apps.size();
			for(int i = 0; i < appCount; i++){
				ObjectNode appObject = (ObjectNode)apps.get(i);
				String name = appObject.get("name").asText();
				if(name.equals("CPS-Topology-"+tenantId)){
					String stateString = appObject.get("status").asText();
					if (stateString.equals("ACTIVE") || stateString.equals("REBALANCING")) {
						return true;
					}
				}
			}
			return false;
		}catch(IOException failedToConnect){
			failedToConnect.printStackTrace();
			throw new CPSException(ErrorCode.SERVER_NOT_FOUND_CODE, "Unable to connect to YARN resource manager URL " + resourceManagerURL);
		}catch (NullPointerException e) {
			// TODO: handle exception
			return false;
		}
	
	}

	
	private String getApplicationId(String tenantId) throws CPSException{

		try{
			String url = resourceManagerURL + "api/v1/topology/summary";
			logger.info("connecting to storm resource manager: " + url);
			String request = HttpClientMonitor.sendGetRequest(url);
			ObjectNode object = JSONUtility.fromObjectString(request);
			ArrayNode apps = (ArrayNode)object.get("topologies");
			int appCount = apps.size();
			for(int i = 0; i < appCount; i++){
				ObjectNode appObject = (ObjectNode)apps.get(i);
				String name = appObject.get("name").asText();
				if(name.equals("CPS-Topology-"+tenantId)){
					String stateString = appObject.get("status").asText();
					AppState state = AppState.valueOf(stateString);
					if(runningStates.contains(state)){
						String applicationId = appObject.get("id").asText();
						return applicationId;
					}
				}
			}
			return null;
		}catch(IOException failedToConnect){
			failedToConnect.printStackTrace();
			throw new CPSException(ErrorCode.SERVER_NOT_FOUND_CODE, "Unable to connect to YARN resource manager URL " + resourceManagerURL);
		}catch (NullPointerException e) {
			// TODO: handle exception
			return null;
		}
	
	}
	
	private static List<AppState> runningStates = Lists.newArrayList(AppState.ACTIVE, AppState.INACTIVE, AppState.REBALANCING, AppState.KILLED);
	
	private static enum AppState {
		ACTIVE,INACTIVE,REBALANCING,KILLED
	}
	
	@Override
	public void registerStarted(String tenantId, WorkerProcess process) throws CPSException {
		
	}
	
//	/api/v1/topology/:id/kill/:wait-time (POST)
//	Kills a topology.
	@Override
	public void stop(String tenantId) throws CPSException {
		String applicationId = getApplicationId(tenantId);
		String url = resourceManagerURL + "/api/v1/topology/" + tenantId + "/kill/6";
		try {
			String response = HttpClientMonitor.sendRequestWithBody(url, "", HTTP_REQUEST_METHOD.post);
		} catch (ClientProtocolException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

}
